/*
 * Copyright 2012 The Netty Project
 *
 * The Netty Project licenses this file to you under the Apache License,
 * version 2.0 (the "License"); you may not use this file except in compliance
 * with the License. You may obtain a copy of the License at:
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations
 * under the License.
 */

package io.netty.buffer;

import java.nio.ByteBuffer;
import java.util.ArrayDeque;
import java.util.Deque;

/**
 * Description of algorithm for PageRun/PoolSubpage allocation from PoolChunk
 *
 * Notation: The following terms are important to understand the code
 * > page  - a page is the smallest unit of memory chunk that can be allocated
 * > chunk - a chunk is a collection of pages
 * > in this code chunkSize = 2^{maxOrder} * pageSize
 *
 * To begin we allocate a byte array of size = chunkSize
 * Whenever a ByteBuf of given size needs to be created we search for the first position
 * in the byte array that has enough empty space to accommodate the requested size and
 * return a (long) handle that encodes this offset information, (this memory segment is then
 * marked as reserved so it is always used by exactly one ByteBuf and no more)
 *
 * For simplicity all sizes are normalized according to PoolArena#normalizeCapacity method
 * This ensures that when we request for memory segments of size >= pageSize the normalizedCapacity
 * equals the next nearest power of 2
 *
 * To search for the first offset in chunk that has at least requested size available we construct a
 * complete balanced binary tree and store it in an array (just like heaps) - memoryMap
 *
 * The tree looks like this (the size of each node being mentioned in the parenthesis)
 *
 * depth=0        1 node (chunkSize)
 * depth=1        2 nodes (chunkSize/2)
 * ..
 * ..
 * depth=d        2^d nodes (chunkSize/2^d)
 * ..
 * depth=maxOrder 2^maxOrder nodes (chunkSize/2^{maxOrder} = pageSize)
 *
 * depth=maxOrder is the last level and the leafs consist of pages
 *
 * With this tree available searching in chunkArray translates like this:
 * To allocate a memory segment of size chunkSize/2^k we search for the first node (from left) at height k
 * which is unused
 *
 * Algorithm:
 * ----------
 * Encode the tree in memoryMap with the notation
 *   memoryMap[id] = x => in the subtree rooted at id, the first node that is free to be allocated
 *   is at depth x (counted from depth=0) i.e., at depths [depth_of_id, x), there is no node that is free
 *
 *  As we allocate & free nodes, we update values stored in memoryMap so that the property is maintained
 *
 * Initialization -
 *   In the beginning we construct the memoryMap array by storing the depth of a node at each node
 *     i.e., memoryMap[id] = depth_of_id
 *
 * Observations:
 * -------------
 * 1) memoryMap[id] = depth_of_id  => it is free / unallocated
 * 2) memoryMap[id] > depth_of_id  => at least one of its child nodes is allocated, so we cannot allocate it, but
 *                                    some of its children can still be allocated based on their availability
 * 3) memoryMap[id] = maxOrder + 1 => the node is fully allocated & thus none of its children can be allocated, it
 *                                    is thus marked as unusable
 *
 * Algorithm: [allocateNode(d) => we want to find the first node (from left) at height h that can be allocated]
 * ----------
 * 1) start at root (i.e., depth = 0 or id = 1)
 * 2) if memoryMap[1] > d => cannot be allocated from this chunk
 * 3) if left node value <= h; we can allocate from left subtree so move to left and repeat until found
 * 4) else try in right subtree
 *
 * Algorithm: [allocateRun(size)]
 * ----------
 * 1) Compute d = log_2(chunkSize/size)
 * 2) Return allocateNode(d)
 *
 * Algorithm: [allocateSubpage(size)]
 * ----------
 * 1) use allocateNode(maxOrder) to find an empty (i.e., unused) leaf (i.e., page)
 * 2) use this handle to construct the PoolSubpage object or if it already exists just call init(normCapacity)
 *    note that this PoolSubpage object is added to subpagesPool in the PoolArena when we init() it
 *
 * Note:
 * -----
 * In the implementation for improving cache coherence,
 * we store 2 pieces of information depth_of_id and x as two byte values in memoryMap and depthMap respectively
 *
 * memoryMap[id]= depth_of_id  is defined above
 * depthMap[id]= x  indicates that the first node which is free to be allocated is at depth x (from root)
 */
final class PoolChunk<T> implements PoolChunkMetric {

    private static final int INTEGER_SIZE_MINUS_ONE = Integer.SIZE - 1;

    /**netty内存池总的数据结构
     *
     */
    final PoolArena<T> arena;
    /**当前申请的内存块，比如对于堆内存，T就是一个byte数组，对于直接内存，T就是ByteBuffer，但无论是哪种形式，其内存大小都默认是16M
     *
     */
    final T memory;
    /**指定当前是否使用内存池的方式进行管理
     *
     */
    final boolean unpooled;
    /**表示当前申请的内存块中有多大一部分是用于站位使用的，整个内存块的大小是16M+offset，默认该值为0
     *
     */
    final int offset;
    /**存储了当前代表内存池的二叉树的各个节点的内存使用情况，该数组长度为4096，二叉树的头结点在该数组的第1号位，存储的值为0；两个一级子节点在该数组的第2号位和3号位，存储的值为1，依次类推。
     * <br/>二叉树的叶节点个数为2048，因而总节点数为4095。在进行内存分配时，会从头结点开始比较，然后比较左子节点，然后比较右子节点，直到找到能够代表目标内存块的节点。
     * <br/>当某个节点所代表的内存被申请之后，该节点的值就会被标记为12，表示该节点已经被占用
     *
     */
    private final byte[] memoryMap;
    /**这里depthMap存储的数据结构与memoryMap是完全一样的，只不过其值在初始化之后一直不会发生变化。该数据的主要作用在于通过目标索引位置值找到其在整棵树中对应的层数
     *
     */
    private final byte[] depthMap;
    /**这里每一个PoolSubPage代表了二叉树的一个叶节点，也就是说，当二叉树叶节点内存被切分成PoolSubPage之后，其会使用一个PoolSubPage对其进行封装
     *
     */
    private final PoolSubpage<T>[] subpages;
    /** Used to determine if the requested capacity is equal to or greater than pageSize. */
    /**其值为-8192，二进制表示为11111111111111111110000000000000，它的后面0的个数正好为12，而2^12=8192，因而将其与用户希望申请的内存大小进行“与操作“，如果其值不为0，
     * <br/>就表示用户希望申请的内存在8192之上，从而就可以快速判断其是在通过PoolSubPage的方式进行申请还是通过内存计算的方式。
     *
     */
    private final int subpageOverflowMask;
    /**记录了每个业节点内存的大小，默认为8192，即8KB
     *
     */
    private final int pageSize;
    /**页节点所代表的偏移量，默认为13，主要作用是计算目标内存在内存池中是在哪个层中，具体的计算公式为：
     * <br/>int d = maxOrder - (log2(normCapacity) - pageShifts); 比如9KB，经过log2(9KB)得到14，maxOrder为11，计算就得到10，表示9KB内存在内存池中为第10层的数据
     *
     */
    private final int pageShifts;
    /**默认为11，表示当前你最大的层数
     *
     */
    private final int maxOrder;
    /**记录了当前整个PoolChunk申请的内存大小，默认为16M
     *
     */
    private final int chunkSize;
    /**将chunkSize取2的对数，默认为24
     *
     */
    private final int log2ChunkSize;
    /**指定了代表叶节点的PoolSubPage数组所需要初始化的长度
     *
     */
    private final int maxSubpageAllocs;
    /** Used to mark memory as unusable */
    /**指定了某个节点如果已经被申请，那么其值将被标记为unusable所指定的值
     *
     */
    private final byte unusable;

    // Use as cache for ByteBuffer created from the memory. These are just duplicates and so are only a container
    // around the memory itself. These are often needed for operations within the Pooled*ByteBuf and so
    // may produce extra GC, which can be greatly reduced by caching the duplicates.
    //
    // This may be null if the PoolChunk is unpooled as pooling the ByteBuffer instances does not make any sense here.
    /**对创建的ByteBuffer进行缓存的一个队列
     *
     */
    private final Deque<ByteBuffer> cachedNioBuffers;

    /**记录了当前PoolChunk中还剩余的可申请的字节数
     *
     */
    private int freeBytes;

    /**在Netty的内存池中，所有的PoolChunk都是由当前PoolChunkList进行组织的
     *
     */
    PoolChunkList<T> parent;
    /**在PoolChunkList中当前PoolChunk的前置节点
     *
     */
    PoolChunk<T> prev;
    /**在PoolChunkList中当前PoolChunk的后置节点
     *
     */
    PoolChunk<T> next;

    // TODO: Test if adding padding helps under contention
    // private long pad0, pad1, pad2, pad3, pad4, pad5, pad6, pad7;
    PoolChunk(PoolArena<T> arena, T memory, int pageSize, int maxOrder, int pageShifts, int chunkSize, int offset) {
        unpooled = false;
        this.arena = arena;
        this.memory = memory;
        this.pageSize = pageSize;
        this.pageShifts = pageShifts;
        this.maxOrder = maxOrder;
        this.chunkSize = chunkSize;
        this.offset = offset;
        unusable = (byte) (maxOrder + 1);
        log2ChunkSize = log2(chunkSize);
        subpageOverflowMask = ~(pageSize - 1);
        freeBytes = chunkSize;

        assert maxOrder < 30 : "maxOrder should be < 30, but is: " + maxOrder;
        maxSubpageAllocs = 1 << maxOrder;

        // Generate the memory map.
        memoryMap = new byte[maxSubpageAllocs << 1];
        depthMap = new byte[memoryMap.length];
        int memoryMapIndex = 1;
        // 首先将参数传入的值进行赋值
    	// this.memory = memory 就是将参数中创建的堆外内存进行保存, 就是chunk所指向的那块连续的内存, 在这个chunk中所分配的ByteBuf, 都会在这块内存中进行读写
    	// 我们重点关注 memoryMap = new byte[maxSubpageAllocs << 1] 和 depthMap = new byte[memoryMap.length] 这两步
    	// 首先看 memoryMap = new byte[maxSubpageAllocs << 1]
    	// 这里初始化了一个字节数组memoryMap, 大小为maxSubpageAllocs << 1, 也就是4096
    	// depthMap = new byte[memoryMap.length] 同样也是初始化了一个字节数组, 大小为memoryMap的大小, 也就是4096
    	// 继续往下分析之前, 我们看chunk的一个层级关系，见《chunk层级结构》
    	// 这是一个二叉树的结构, 左侧的数字代表层级, 右侧代表一块连续的内存, 每个父节点下又拆分成多个子节点, 最顶层表示的内存范围为0-16MB, 其又下分为两层, 范围为0-8MB, 8-16MB, 以此类推, 最后到11层, 以8k的大小划分, 也就是一个page的大小
    	// 如果我们分配一个8mb的缓冲区, 则会将第二层的第一个节点, 也就是0-8这个连续的内存进行分配, 分配完成之后, 会将这个节点设置为不可用, 具体逻辑后面会讲解
    	// 结合《chunk层级结构》图, 我们再看构造方法中的for循环:
        // 实际上这个for循环就是将上面的结构包装成一个字节数组memoryMap, 外层循环用于控制层数, 内层循环用于控制里面每层的节点, 这里经过循环之后, memoryMap和depthMap内容为以下表现形式:
    	// [0, 0, 1, 1, 2, 2, 2, 2, 3, 3, 3, 3, 3, 3, 3, 3, 4, 4, 4, 4, 4, 4, 4, 4, 4, 4, 4, 4, 4, 4, 4, 4...........]
        // 这里注意一下, 因为程序中数组的下标是从1开始设置的, 所以第零个节点元素为默认值0
    	// 这里数字代表层级, 同时也代表了当前层级的节点, 相同的数字个数就是这一层级的节点数
    	// 其中0为2个(因为这里分配时下标是从1开始的, 所以第0个位置是默认值0, 实际上第零层元素只有一个, 就是头结点), 1为2个, 2为4个, 3为8个, 4为16个, n为2的n次方个, 直到11, 也就是11有2的11次方个

        // d相当于一个深度, 赋值的内容代表当前节点的深度
        for (int d = 0; d <= maxOrder; ++ d) { // move down the tree one level at a time
            int depth = 1 << d;
            for (int p = 0; p < depth; ++ p) {
                // in each level traverse left to right and set value to the depth of subtree
                memoryMap[memoryMapIndex] = (byte) d;
                depthMap[memoryMapIndex] = (byte) d;
                memoryMapIndex ++;
            }
        }

        subpages = newSubpageArray(maxSubpageAllocs);
        // cachedNioBuffers: 对创建的ByteBuffer进行缓存的一个队列
        cachedNioBuffers = new ArrayDeque<ByteBuffer>(8);
    }

    /** Creates a special chunk that is not pooled. */
    PoolChunk(PoolArena<T> arena, T memory, int size, int offset) {
        unpooled = true;
        this.arena = arena;
        this.memory = memory;
        this.offset = offset;
        memoryMap = null;
        depthMap = null;
        subpages = null;
        subpageOverflowMask = 0;
        pageSize = 0;
        pageShifts = 0;
        maxOrder = 0;
        unusable = (byte) (maxOrder + 1);
        chunkSize = size;
        log2ChunkSize = log2(chunkSize);
        maxSubpageAllocs = 0;
        cachedNioBuffers = null;
    }

    @SuppressWarnings("unchecked")
    private PoolSubpage<T>[] newSubpageArray(int size) {
        return new PoolSubpage[size];
    }

    @Override
    public int usage() {
        final int freeBytes;
        synchronized (arena) {
            freeBytes = this.freeBytes;
        }
        return usage(freeBytes);
    }

    private int usage(int freeBytes) {
        if (freeBytes == 0) {
            return 100;
        }

        int freePercentage = (int) (freeBytes * 100L / chunkSize);
        if (freePercentage == 0) {
            return 99;
        }
        return 100 - freePercentage;
    }

    boolean allocate(PooledByteBuf<T> buf, int reqCapacity, int normCapacity) {
        final long handle;
        if ((normCapacity & subpageOverflowMask) != 0) { // >= pageSize
        	// Chunk中包含了多个Page，每个Page的大小为8k，如果要分配16k的内存，则在Chunk中找到连续的两个Page就可以
        	// 这里返回的就是memoryMap的一个下标，通过这个下标，我们就能唯一定位一块内存
            handle = allocateRun(normCapacity);
        } else {
        	// 有些时候分配8k也是一种浪费，比如只需要2k的缓冲区，这时，netty会将Page切分成多个SubPage，每个大小要根据分配的缓冲区大小而定，比如要分配2k，就会将一个Page切分为4个SubPage，每个大小为2k
            // 这里返回的handle就指向chunk中的某个page中的某个子块所对应的连续内存
        	handle = allocateSubpage(normCapacity);
        }

        if (handle < 0) {
            return false;
        }
        ByteBuffer nioBuffer = cachedNioBuffers != null ? cachedNioBuffers.pollLast() : null;
        initBuf(buf, nioBuffer, handle, reqCapacity);
        return true;
    }

    /**
     * Update method used by allocate
     * This is triggered only when a successor is allocated and all its predecessors
     * need to update their state
     * The minimal depth at which subtree rooted at id has some free space
     *
     * @param id id
     */
    private void updateParentsAlloc(int id) {
        while (id > 1) {
        	// 取到当前节点的父节点的id
            int parentId = id >>> 1;
            // 获取当前节点的值
            byte val1 = value(id);
            // 找到当前节点的兄弟节点
            byte val2 = value(id ^ 1);
            // 如果当前节点值小于兄弟节点，则保存当前节点值到val，否则，保存兄弟节点值到val
	        // 如果当前节点是不可用，则当前节点值是12，大于兄弟节点的值，所以这里将兄弟节点的值进行保存
            byte val = val1 < val2 ? val1 : val2;
            // 将val的值设置为父节点下标所对应的值
            setValue(parentId, val);
            // id设置为父节点id，继续循环
            id = parentId;
        }
    }

    /**
     * Update method used by free
     * This needs to handle the special case when both children are completely free
     * in which case parent be directly allocated on request of size = child-size * 2
     *
     * @param id id
     */
    private void updateParentsFree(int id) {
        int logChild = depth(id) + 1;
        while (id > 1) {
            int parentId = id >>> 1;
            byte val1 = value(id);
            byte val2 = value(id ^ 1);
            logChild -= 1; // in first iteration equals log, subsequently reduce 1 from logChild as we traverse up

            if (val1 == logChild && val2 == logChild) {
                setValue(parentId, (byte) (logChild - 1));
            } else {
                byte val = val1 < val2 ? val1 : val2;
                setValue(parentId, val);
            }

            id = parentId;
        }
    }

    /**从第一个节点开始往下找，找到层级为d、未被使用的节点，找到后通过setValue将当前节点设置为不可用，其中id是当前节点的下标，unusable代表一个不可用的值，这里是12，
     * <br/>因为我们的层级只有12层，所以设置为12后相当于标记不可用。设置为不可用后，通过updateParentsAlloc逐层设置为缓存被使用的状态
     * <br/>Algorithm to allocate an index in memoryMap when we query for a free node at depth d
     *
     * @param d depth
     * @return index in memoryMap
     */
    private int allocateNode(int d) {
    	// 下标初始值为1
        int id = 1;
        // 代表当前层级第一个节点的初始下标
        int initial = - (1 << d); // has last d bits = 0 and rest all = 1
        // 获取第一个节点的值
        byte val = value(id);
        // 如果值大于层级，说明chunk不可用
        if (val > d) { // unusable
            return -1;
        }
        // 当前下标对应的节点值如果小于层级，或者当前下标小于层级的初始下标
        while (val < d || (id & initial) == 0) { // id & initial == 1 << d for all ids at depth d, for < d it is 0
        	// 当前下标乘以2，代表当前节点的子节点的起始位置
        	id <<= 1;
        	// 获得id位置的值
            val = value(id);
            // 如果当前节点值大于层数(节点不可用)
            if (val > d) {
            	// id为偶数则+1，id为奇数则-1（拿的是其兄弟节点）
                id ^= 1;
                // 获取id的值
                val = value(id);
            }
        }
        byte value = value(id);
        assert value == d && (id & initial) == 1 << d : String.format("val = %d, id & initial = %d, d = %d", value, id & initial, d);
        // 将找到的节点设置为不可用
        setValue(id, unusable); // mark as unusable
        // 逐层往上标记被使用
        updateParentsAlloc(id);
        return id;
    }

    /**申请的大于等于1个page的空间。
     * <br/>Chunk中包含了多个Page，每个Page的大小为8k，如果要分配16k的内存，则在Chunk中找到连续的两个Page就可以
     * <br/>Allocate a run of pages (>=1)
     *
     * @param normCapacity normalized capacity
     * @return index in memoryMap memoryMap的索引
     */
    private long allocateRun(int normCapacity) {
    	// 根据normCapacity计算出第几层
        int d = maxOrder - (log2(normCapacity) - pageShifts); // 比如9KB，经过log2(9KB)得到14，maxOrder为11，计算就得到10，表示9KB内存在内存池中为第10层的数据
        // 根据层级关系去分配一个节点，返回的id是memoryMap的下标
        int id = allocateNode(d);
        if (id < 0) {
            return id;
        }
        freeBytes -= runLength(id);
        return id;
    }

    /**PoolSubpage数组有点类似于SubPage的缓存，我们创建了一个SubPage之后，会将创建好的SubPage与PoolArena中的tinySubpagePools数据的每一个元素进行关联，
     * <br/>下次再分配的时候可以直接通过tinySubpagePools数组元素去找关联的SubPage
     * <br/>Create / initialize a new PoolSubpage of normCapacity
     * Any PoolSubpage created / initialized here is added to subpage pool in the PoolArena that owns this PoolChunk
     *
     * @param normCapacity normalized capacity
     * @return index in memoryMap
     */
    private long allocateSubpage(int normCapacity) {
        // Obtain the head of the PoolSubPage pool that is owned by the PoolArena and synchronize on it.
        // This is need as we may add it back and so alter the linked-list structure.
    	// 获得对应的内存级别数组的head节点
    	PoolSubpage<T> head = arena.findSubpagePoolHead(normCapacity);
        // 表示在第11层分配节点，maxOrder: 11
        int d = maxOrder; // subpages are only be allocated from pages i.e., leaves
        synchronized (head) {
        	// 获取字节数组memoryMap的下标，这里指向一个Page，如果是第一次分配，指向的是0-8k的那个Page
            int id = allocateNode(d);
            if (id < 0) {
                return id;
            }

            // 获取初始化的subpage
 			// 是拿到PoolChunk中成员变量subpages的值, 也是个PoolSubpage的数组, 在PoolChunk进行初始化的时候, 也会初始化该数组, 长度为2048
 			// 也就是说每个chunk都维护着一个subpage的列表, 如果每一个page级别的内存都需要被切分成子page, 则会将这个这个page放入该列表中, 专门用于分配子page, 所以这个列表中的subpage, 其实就是一个用于切分的page
            final PoolSubpage<T>[] subpages = this.subpages;
            final int pageSize = this.pageSize;

            freeBytes -= pageSize;

            int subpageIdx = subpageIdx(id);
            PoolSubpage<T> subpage = subpages[subpageIdx];
            // 因为默认的subpages只是创建一个数组，并没有往数组中赋值，所以第一次执行到这里会返回true
            if (subpage == null) {
            	// 这里通过new PoolSubpage创建一个新的subpage之后, 通过subpages[subpageIdx] = subpage这种方式将新创建的subpage根据下标赋值到subpages中的元素中
                subpage = new PoolSubpage<T>(head, this, id, runOffset(id), pageSize, normCapacity);
                subpages[subpageIdx] = subpage;
            } else {
                subpage.init(head, normCapacity);
            }
            return subpage.allocate();
        }
    }

    /**
     * Free a subpage or a run of pages
     * When a subpage is freed from PoolSubpage, it might be added back to subpage pool of the owning PoolArena
     * If the subpage pool in PoolArena has at least one other PoolSubpage of given elemSize, we can
     * completely free the owning Page so it is available for subsequent allocations
     *
     * @param handle handle to free
     */
    void free(long handle, ByteBuffer nioBuffer) {
        int memoryMapIdx = memoryMapIdx(handle);
        int bitmapIdx = bitmapIdx(handle);

        // bitmapIdx != 0判断当前缓冲区分配的级别是Page还是SubPage
        if (bitmapIdx != 0) { // free a subpage
        	// 如果是SubPage，则会找到相关的SubPage将其位图标记为0
            PoolSubpage<T> subpage = subpages[subpageIdx(memoryMapIdx)];
            assert subpage != null && subpage.doNotDestroy;

            // Obtain the head of the PoolSubPage pool that is owned by the PoolArena and synchronize on it.
            // This is need as we may add it back and so alter the linked-list structure.
            // 获取PoolArena拥有的PoolSubPage池头并对其进行同步。这是需要的，因为我们可以把它添加回去，从而改变链表结构
            PoolSubpage<T> head = arena.findSubpagePoolHead(subpage.elemSize);
            synchronized (head) {
            	// 返回true，说明该Chunk还在使用；否则就可以释放该Chunk
                if (subpage.free(head, bitmapIdx & 0x3FFFFFFF)) {
                    return;
                }
            }
        }
        // 通过分配内存的反向标记，将该内存标记为未使用
        freeBytes += runLength(memoryMapIdx);
        setValue(memoryMapIdx, depth(memoryMapIdx));
        updateParentsFree(memoryMapIdx);

        if (nioBuffer != null && cachedNioBuffers != null &&
                cachedNioBuffers.size() < PooledByteBufAllocator.DEFAULT_MAX_CACHED_BYTEBUFFERS_PER_CHUNK) {
            cachedNioBuffers.offer(nioBuffer);
        }
    }

    void initBuf(PooledByteBuf<T> buf, ByteBuffer nioBuffer, long handle, int reqCapacity) {
    	// 这里通过memoryMapIdx(handle)找到memoryMap的下标，其实就是handle的int值
        int memoryMapIdx = memoryMapIdx(handle);
        // bitmapIdx是关于SubPage中使用到的逻辑，如果是normal（Page）级别的分配，只返回0
        int bitmapIdx = bitmapIdx(handle);
        if (bitmapIdx == 0) {
            byte val = value(memoryMapIdx);
            // 判断当前节点是不是不可用状态，然后通过init方法进行初始化
            assert val == unusable : String.valueOf(val);
            buf.init(this, nioBuffer, handle, runOffset(memoryMapIdx) + offset, reqCapacity, runLength(memoryMapIdx), arena.parent.threadCache());
        } else {
            initBufWithSubpage(buf, nioBuffer, handle, bitmapIdx, reqCapacity);
        }
    }

    void initBufWithSubpage(PooledByteBuf<T> buf, ByteBuffer nioBuffer, long handle, int reqCapacity) {
        initBufWithSubpage(buf, nioBuffer, handle, bitmapIdx(handle), reqCapacity);
    }

    private void initBufWithSubpage(PooledByteBuf<T> buf, ByteBuffer nioBuffer, long handle, int bitmapIdx, int reqCapacity) {
        assert bitmapIdx != 0;
        int memoryMapIdx = memoryMapIdx(handle);
        PoolSubpage<T> subpage = subpages[subpageIdx(memoryMapIdx)];
        assert subpage.doNotDestroy;
        assert reqCapacity <= subpage.elemSize;

        buf.init(
            this, nioBuffer, handle,
            // 我们注意在buf调用init方法中的一个参数:  runOffset(memoryMapIdx) + (bitmapIdx & 0x3FFFFFFF) * subpage.elemSize
			// 这里的偏移量就是, 原来page的偏移量+子块的偏移量
			// bitmapIdx & 0x3FFFFFFF 代表当前分配的子page是属于第几个子page
			// (bitmapIdx & 0x3FFFFFFF) * subpage.elemSize  表示在当前page的偏移量
			// 这样, 分配的ByteBuf在内存读写的时候, 就会根据偏移量进行读写
            runOffset(memoryMapIdx) + (bitmapIdx & 0x3FFFFFFF) * subpage.elemSize + offset,
                reqCapacity, subpage.elemSize, arena.parent.threadCache());
    }

    private byte value(int id) {
        return memoryMap[id];
    }

    private void setValue(int id, byte val) {
        memoryMap[id] = val;
    }

    private byte depth(int id) {
        return depthMap[id];
    }

    private static int log2(int val) {
        // compute the (0-based, with lsb = 0) position of highest set bit i.e, log2
        return INTEGER_SIZE_MINUS_ONE - Integer.numberOfLeadingZeros(val);
    }

    /**表示根据下标获取可分配的最大长度
     * @param id
     * @return
     */
    private int runLength(int id) {
        // represents the size in #bytes supported by node 'id' in the tree
        return 1 << log2ChunkSize - depth(id);
    }

    /**表示偏移量，相当于分配给缓冲区的这块内存相对于Chunk中申请的内存的首地址偏移了多少
     * @param id
     * @return
     */
    private int runOffset(int id) {
        // represents the 0-based offset in #bytes from start of the byte-array chunk
        int shift = id ^ 1 << depth(id);
        return shift * runLength(id);
    }

    private int subpageIdx(int memoryMapIdx) {
    	// 如果id对应的Page是0-8k的节点，这里获得的下标就是0
        return memoryMapIdx ^ maxSubpageAllocs; // remove highest set bit, to get offset
    }

    private static int memoryMapIdx(long handle) {
        return (int) handle;
    }

    private static int bitmapIdx(long handle) {
        return (int) (handle >>> Integer.SIZE);
    }

    @Override
    public int chunkSize() {
        return chunkSize;
    }

    @Override
    public int freeBytes() {
        synchronized (arena) {
            return freeBytes;
        }
    }

    @Override
    public String toString() {
        final int freeBytes;
        synchronized (arena) {
            freeBytes = this.freeBytes;
        }

        return new StringBuilder()
                .append("Chunk(")
                .append(Integer.toHexString(System.identityHashCode(this)))
                .append(": ")
                .append(usage(freeBytes))
                .append("%, ")
                .append(chunkSize - freeBytes)
                .append('/')
                .append(chunkSize)
                .append(')')
                .toString();
    }

    void destroy() {
        arena.destroyChunk(this);
    }
}
